package com.boot.notes.rocketmq;

import cn.hutool.json.JSONUtil;
import com.boot.common.utils.R;
import com.boot.modules.admin.form.SysLoginForm;
import com.boot.notes.event.OrderEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.List;
import java.util.Map;

@Slf4j
@Controller
@RequestMapping("/notes/mq")
public class MyProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    @GetMapping("/send")
    public R send(SendCallback sendCallback) throws IOException, MQClientException {


        // 如下两种方式等价
        rocketMQTemplate.convertAndSend("test-topic-1", "我就是个字符串");
        rocketMQTemplate.convertAndSend("test-topic-1", new OrderBean("自定义 OrderBean 对象", 1l)); //自定义对象 如果消费者是用string接收，则会自动转为json串
        rocketMQTemplate.send("test-topic-1", MessageBuilder.withPayload(new OrderBean("8888888888", 2l)).build());


        /**
         * 发送同步消息
         *
         * 发送消息采用同步模式，这种方式只有在消息完全发送完成之后才返回结果，此方式存在需要同步等待发送结果的时间代价。
         * 这种方式具有内部重试机制，即在主动声明本次消息发送失败之前，内部实现将重试一定次数，默认为2次（DefaultMQProducer＃getRetryTimesWhenSendFailed）。
         * 发送的结果存在同一个消息可能被多次发送给给broker，这里需要应用的开发者自己在消费端处理幂等性问题。
         *
         */
        rocketMQTemplate.syncSend("test-topic-1", "同步模式：syncSend 此方式存在需要同步等待发送结果的时间代价，这种方式具有内部重试机制，默认为2次", Long.parseLong("1812281114"));


        /**
         * 发送带tag的消息，格式：topic:tag，示例：test-topic-1:paid
         *
         **/
        rocketMQTemplate.convertAndSend("test-topic-1:paid", "test-topic-1:paid");

        // 发送延迟消息
        /**
         * 延时消息等级分为18个：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
         * 延时时间不能自定义，自定义巴巴是要收费的
         * @return com.boot.common.utils.R
         **/
        rocketMQTemplate.asyncSend("test-topic-1", MessageBuilder.withPayload("我是延时消息").build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步消息 send successful" + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.info("send fail; {}", e.getMessage());
            }

        }, 3000, 3);

        // 发送即发即失消息（不关心发送结果）
        /**
         * 采用one-way发送模式发送消息的时候，发送端发送完消息后会立即返回，不会等待来自broker的ack来告知本次消息发送是否完全完成发送。
         * 这种方式吞吐量很大，但是存在消息丢失的风险，所以其适用于不重要的消息发送，比如日志收集。
         *
         * @Description:
         * @Author: mkq
         * @Date: 2021/10/11 15:25

         * @return com.boot.common.utils.R
         **/
        rocketMQTemplate.sendOneWay("test-topic-1", MessageBuilder.withPayload("I'm one way message").build());


        // 发送顺序消息
        rocketMQTemplate.syncSendOrderly("test-topic-4", "I'm order message", "1234");


        /**
         * 发送异步消息
         *
         * 发送消息采用异步发送模式，消息发送后立刻返回，当消息完全完成发送后，会调用回调函数sendCallback来告知发送者本次发送是成功或者失败。
         * 异步模式通常用于响应时间敏感业务场景，即承受不了同步发送消息时等待返回的耗时代价。
         *
         * 同同步发送一样，异步模式也在内部实现了重试机制，默认次数为2次（DefaultMQProducer#getRetryTimesWhenSendAsyncFailed}）。
         * 发送的结果同样存在同一个消息可能被多次发送给给broker，需要应用的开发者自己在消费端处理幂等性问题。
         *
         **/
        rocketMQTemplate.asyncSend("test-topic-2", MessageBuilder.withPayload("我是异步消息！！！").build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步消息 send successful" + sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.info("send fail; {}", e.getMessage());
            }
        });

        System.out.println("send finished!");


        return R.ok();
    }


    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        DefaultMQProducer producer = new  DefaultMQProducer("my-producer-group-111");
        producer.setNamesrvAddr("192.168.0.166:9876");
        // 订阅 Topic
        producer.setCreateTopicKey("test-topic-1");
        Message msg = new Message();
        msg.setTopic("test-topic-1");
        msg.setTags("ppp");
        msg.setBody(JSONUtil.toJsonStr(new OrderBean("ta123", 1l)).getBytes());
        producer.start();
        producer.send(msg);
    }



}
